#include "StdAfx.h"
#include "QueryIntermediateRepresentationInterpreter.h"
#include "Types.h"
#include "RootOperator.h"
#include "IStreamOutput.h"
#include "QueryManager.h"
#include "DstreamOperator.h"
#include "IstreamOperator.h"
#include "RstreamOperator.h"
#include "ProjectionOperator.h"
#include "SelectionOperator.h"
#include "JoinOperator.h"
#include "RowWindowOperator.h"
#include "RangeWindowOperator.h"
#include "LeafOperator.h"
#include <boost/shared_ptr.hpp>
QueryIntermediateRepresentationInterpreter* QueryIntermediateRepresentationInterpreter::queryIntermediateRepresentationInterpreter=NULL;
QueryIntermediateRepresentationInterpreter::QueryIntermediateRepresentationInterpreter()
{

}


QueryIntermediateRepresentationInterpreter::~QueryIntermediateRepresentationInterpreter(void)
{
}
QueryIntermediateRepresentationInterpreter* QueryIntermediateRepresentationInterpreter::getInstance()
{
	if(queryIntermediateRepresentationInterpreter==NULL)
	{
		queryIntermediateRepresentationInterpreter = new QueryIntermediateRepresentationInterpreter();
	}
	return queryIntermediateRepresentationInterpreter;
}
boost::shared_ptr<Operator> QueryIntermediateRepresentationInterpreter::resolve(Document& document)
{
	std::string operatorKind = document.getField(QUERY_TYPE).valuestr();
	 
	if(operatorKind==QUERY_ROOT)
	{
		boost::shared_ptr<RootOperator> rootOperator (new RootOperator());
		boost::shared_ptr<IStreamOutput> streamOutput = QueryManager::getInstance()->getStreamOutput();
		rootOperator->setStreamOutput(streamOutput);
		Document& inputDocument = document.getField(QUERY_INPUT).embeddedObject();
		//std::cout<<inputDocument<<std::endl;
		boost::shared_ptr<Operator> inputOperator = resolve(inputDocument);
		connectTwoOperator(inputOperator,rootOperator);
		return rootOperator;
	}
	else if(operatorKind==QUERY_ISTREAM)
	{
		boost::shared_ptr<IstreamOperator> istreamOperator (new IstreamOperator());
		Document& inputDocument = document.getField(QUERY_INPUT).embeddedObject();
		boost::shared_ptr<Operator> inputOperator = resolve(inputDocument);
		connectTwoOperator(inputOperator,istreamOperator);
		return istreamOperator;
	}
	else if(operatorKind==QUERY_DSTREAM)
	{
		boost::shared_ptr<DstreamOperator> dstreamOperator (new DstreamOperator());
		Document& inputDocument = document.getField(QUERY_INPUT).embeddedObject();
		boost::shared_ptr<Operator> inputOperator = resolve(inputDocument);
		connectTwoOperator(inputOperator,dstreamOperator);
		return dstreamOperator;
	}
	else if(operatorKind==QUERY_RSTREAM)
	{
		boost::shared_ptr<RstreamOperator> rstreamOperator (new RstreamOperator());
		Document& inputDocument = document.getField(QUERY_INPUT).embeddedObject();
		boost::shared_ptr<Operator> inputOperator = resolve(inputDocument);
		connectTwoOperator(inputOperator,rstreamOperator);
		return rstreamOperator;
	}
	else if(operatorKind==QUERY_PROJECTION)
	{
		Document & projectionAttributeDocument = document.getField(QUERY_PROJECTION_ATTRIBUTE_LIST).embeddedObject();
		//std::cout<<projectionAttributeDocument<<std::endl;
		std::set<std::string> projectionAttributeSet;
		DocumentIterator it(projectionAttributeDocument);
		while(it.more())
		{
			DocumentElement& documentElement =  it.next();	
			//std::cout<<documentElement<<std::endl;
			std::string projectionAttribute = documentElement.valuestr();
			projectionAttributeSet.insert(projectionAttribute);
		}
		boost::shared_ptr<ProjectionOperator> projectionOperator (new ProjectionOperator(projectionAttributeSet));
	
		Document& inputDocument = document.getField(QUERY_INPUT).embeddedObject();
		boost::shared_ptr<Operator> inputOperator = resolve(inputDocument);
		connectTwoOperator(inputOperator,projectionOperator);
		return projectionOperator;
	}
	else if(operatorKind==QUERY_SELECTION)
	{
		boost::shared_ptr<SelectionOperator> selectionOperator (new SelectionOperator());
		std::string selectionCondition = document.getField(QUERY_SELECTION_CONDITION).valuestr();
		selectionOperator->setSelectionCondition(selectionCondition);
		Document& inputDocument = document.getField(QUERY_INPUT).embeddedObject();
		boost::shared_ptr<Operator> inputOperator = resolve(inputDocument);
		connectTwoOperator(inputOperator,selectionOperator);
		return selectionOperator;
	}
	else if(operatorKind==QUERY_JOIN)
	{
		boost::shared_ptr<JoinOperator> joinOperator (new JoinOperator());
		std::string leftJoinAttribute = document.getField(QUERY_LEFT_JOIN_ATTRIBUTE).valuestr();
		std::string rightJoinAttribute = document.getField(QUERY_RIGHT_JOIN_ATTRIBUTE).valuestr();
		joinOperator->setLeftJoinAttribute(leftJoinAttribute);
		joinOperator->setRightJoinAttribute(rightJoinAttribute);
		Document& leftInputDocument = document.getField(QUERY_LEFT_INPUT).embeddedObject();
		Document& rightInputDocument = document.getField(QUERY_RIGHT_INPUT).embeddedObject();
		boost::shared_ptr<Operator> leftInputOperator = resolve(leftInputDocument);
		boost::shared_ptr<Operator> rightInputOperator = resolve(rightInputDocument);
		connectThreeOperator(leftInputOperator,rightInputOperator,joinOperator);
		return joinOperator;
	}
	else if(operatorKind==QUERY_ROWWINDOW)
	{
		//std::cout<<document.getField(QUERY_WINDOW_SIZE)<<std::endl;
		int rowWindowSize = document.getField(QUERY_WINDOW_SIZE).numberInt();
		boost::shared_ptr<RowWindowOperator> rowWindowOperator (new RowWindowOperator(rowWindowSize));

		Document& inputDocument = document.getField(QUERY_INPUT).embeddedObject();
		boost::shared_ptr<Operator> inputOperator = resolve(inputDocument);
		connectTwoOperator(inputOperator,rowWindowOperator);
		return rowWindowOperator;
	}
	else if(operatorKind==QUERY_RANGEWINDOW)
	{
		Timeunit timeRange = document.getField(QUERY_WINDOW_SIZE).Long();
		boost::shared_ptr<RangeWindowOperator> rangeWindowOperator (new RangeWindowOperator(timeRange));

		Document& inputDocument = document.getField(QUERY_INPUT).embeddedObject();
		boost::shared_ptr<Operator> inputOperator = resolve(inputDocument);
		connectTwoOperator(inputOperator,rangeWindowOperator);
		return rangeWindowOperator;
	}
	else if(operatorKind==QUERY_LEAF)
	{
		boost::shared_ptr<LeafOperator> leafOperator (new LeafOperator());
		std::string sourceId = document.getField(QUERY_STREAM_SOURCE).valuestr();
		boost::shared_ptr<IStreamInput> streamInput = QueryManager::getInstance()->getStreamInputById(sourceId);
		leafOperator->setStreamInput(streamInput);
		return leafOperator;
	}
	else
	{
		assert(false);//never reached
	}
}

void QueryIntermediateRepresentationInterpreter::connectTwoOperator(boost::shared_ptr<Operator>inputOperator,boost::shared_ptr<Operator>outputOperator)
{
	boost::shared_ptr<QueueEntity>queueEntity(new QueueEntity());
	inputOperator->addOutputQueue(queueEntity);
	queueEntity->setInputOperator(inputOperator);
	outputOperator->addInputQueue(queueEntity);
	queueEntity->setOutputOperator(outputOperator);
}
void QueryIntermediateRepresentationInterpreter::connectThreeOperator(boost::shared_ptr<Operator>leftInputOperator,boost::shared_ptr<Operator>rightInputOperator,boost::shared_ptr<Operator>outputOperator)
{
	boost::shared_ptr<QueueEntity>leftQueueEntity(new QueueEntity());
	leftInputOperator->addOutputQueue(leftQueueEntity);
	leftQueueEntity->setInputOperator(leftInputOperator);
	outputOperator->addInputQueue(leftQueueEntity);
	leftQueueEntity->setOutputOperator(outputOperator);
	boost::shared_ptr<QueueEntity>rightQueueEntity(new QueueEntity());
	rightInputOperator->addOutputQueue(rightQueueEntity);
	rightQueueEntity->setInputOperator(rightInputOperator);
	outputOperator->addInputQueue(rightQueueEntity);
	rightQueueEntity->setOutputOperator(outputOperator);
}